package com.zyb.dfs.datanode.server;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * 负责解析请求以及发送响应的线程
 */
public class NioProcessor extends Thread {
    /**
     * 多路复用监听最大的等待时间
     */
    private static final Long POLL_BLOCK_MAX_TIME = 1000L;

    /**
     * 等待注册的网络连接队列
     */
    private ConcurrentLinkedDeque<SocketChannel> channels = new ConcurrentLinkedDeque<SocketChannel>();

    // 暂存起来的响应缓存在这里
    private Map<String, NetworkResponse> cachedResponses = new HashMap<String, NetworkResponse>();

    // 这个processor负责维护的所有客户端的SelectionKey
    private Map<String, SelectionKey> cachedKeys = new HashMap<String, SelectionKey>();
    /**
     * 每个processor都有自己私有的processor
     */
    private Selector selector;

    // 标识
    private Integer processorId;

    // 没读取完的请求缓存在这里
    private Map<String, NetworkRequest> cachedRequests =
            new HashMap<String, NetworkRequest>();

    public NioProcessor(Integer processorId) {
        try {
            this.processorId = processorId;
            this.selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public Integer getProcessorId() {
        return processorId;
    }

    public void setProcessorId(Integer processorId) {
        this.processorId = processorId;
    }

    /**
     * 给这个processor分配网络连接
     *
     * @param channel
     */
    public void addChannel(SocketChannel channel) {
        channels.offer(channel);
        // 直接唤醒阻塞的selector（int keys = selector.select(POLL_BLOCK_MAX_TIME);）
        selector.wakeup();
    }

    @Override
    public void run() {
        while (true) {
            try {
                // 注册排队等待的连接
                registerQueuedClients();
                // 处理排队中的响应
                cacheQueuedResponse();
                // 以限时阻塞的方式感知连接中的请求
                poll();

            } catch (Exception e) {
                e.getStackTrace();
            }
        }
    }

    /**
     * 将等待的channel 注册到selector中去
     */
    private void registerQueuedClients() {
        SocketChannel channel = null;
        while ((channel = channels.poll()) != null) {
            try {
                channel.register(selector, SelectionKey.OP_READ);
            } catch (Exception e) {
                e.getStackTrace();
            }
        }
    }

    /**
     * 以多路复用的方式来监听连接的请求
     */
    private void poll() {
        try {
            int keys = selector.select(POLL_BLOCK_MAX_TIME);
            if (keys > 0) {
                Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();
                while (selectionKeys.hasNext()) {
                    SelectionKey key = selectionKeys.next();
                    selectionKeys.remove();

                    SocketChannel channel = (SocketChannel) key.channel();
                    String client = channel.getRemoteAddress().toString();
                    // 接受某个客户端的请求
                    if (key.isReadable()) {

                        NetworkRequest request = null;
                        if (cachedRequests.get(client) != null) {
                            request = cachedRequests.get(client);
                        } else {
                            request = new NetworkRequest();
                        }

                        request.setChannel(channel);
                        request.setKey(key);
                        request.read();

                        if (request.hasCompletedRead()) {
                            // 此时就可以将一个请求分发到全局的请求队列里去了
                            request.setProcessorId(processorId);
                            request.setClient(client);
                            NetworkRequestQueue.get().offer(request);
                            cachedKeys.put(client, key);
                            cachedRequests.remove(client);
                            key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
                        } else {
                            cachedRequests.put(client, request);
                        }
                    } else if (key.isWritable()) {
                        NetworkResponse response = cachedResponses.get(client);
                        channel.write(response.getBuffer());

                        cachedResponses.remove(client);
                        cachedKeys.remove(client);

                        key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                    }
                }
            }
        } catch (Exception e) {

        }
    }

    /**
     * 暂存排队中的响应
     */
    private void cacheQueuedResponse() {
        NetworkResponseQueues responseQueues = NetworkResponseQueues.get();
        NetworkResponse response = null;

        while ((response = responseQueues.poll(processorId)) != null) {
            String client = response.getClient();
            cachedResponses.put(client, response);
            cachedKeys.get(client).interestOps(SelectionKey.OP_WRITE);
        }
    }
}
